package org.luxor.common.netty.jdk.nio.client;

import org.luxor.common.netty.jdk.nio.server.MultiplexerTimeServerHandler;
import org.luxor.commons.core.utils.StringUtils;
import org.mockito.internal.util.io.IOUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;

public class MultiplexerTimeClientHandler implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(MultiplexerTimeServerHandler.class);
    private static final int BUFFER_SIZE = 512;
    private String host;
    private int port;
    private Selector selector;
    private SocketChannel socketChannel;
    private volatile boolean stop;

    /**
     * 初始化多路复用器，绑定监听端口
     */
    public MultiplexerTimeClientHandler(String host, int port) {
        this.host = StringUtils.isBlank(host) ? "127.0.0.1" : host;
        this.port = port;
        try {
            // 创建多路复用器
            selector = Selector.open();
            // 1、打开SocketChannel，绑定客户端本地地址(可选，默认系统会随机分配一个可用的本地地址)
            socketChannel = SocketChannel.open();
            // 2、设置连接为非阻塞模式，同时设置客户端连接的TCP参数
            socketChannel.configureBlocking(false);
            Socket socket = socketChannel.socket();
            socket.setReuseAddress(true);
            socket.setReceiveBufferSize(BUFFER_SIZE);
            socket.setSendBufferSize(BUFFER_SIZE);
            System.out.println("The time client is start in host:" + host + "/ port: " + port);
        } catch (Exception e) {
            log.error("初始化多路复用器,绑定客户端本地地址.失败!", e);
            System.exit(1);
        }
    }

    public void stop() {
        this.stop = true;
    }

    public void closeAll() {
        stop();
        // 多路复用器关闭后，所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭，所以不需要重复释放资源
        IOUtil.close(this.selector);
    }

    @Override
    public void run() {
        try {
            doConnect();
        } catch (Exception e) {
            log.error("创建客户端连接.失败!", e);
            System.exit(1);
        }
        while (!stop) {
            try {
                // 设置多路复用器，检查通道就绪状态的最长等待时间(单位毫秒)
                selector.select(1000);

                // 5、无限循环处理多路复用器中,准备就绪的key-channel
                Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                while (it.hasNext()) {
                    SelectionKey key = it.next();
                    try {
                        inputHandle(key);
                    } catch (Exception e) {
                        if (key != null) key.cancel();
                        if (key.channel() != null) key.channel().close();
                    } finally {
                        it.remove();
                    }
                }
            } catch (Exception e) {
                log.warn("多路复用器，循环处理准备就绪的key时，出现异常", e);
            }
        }
    }

    private void inputHandle(SelectionKey key) throws IOException {
        if (key.isValid()) {
            SocketChannel sc = (SocketChannel) key.channel();
            // 判断是否连接成功
            if (key.isConnectable()) {
                if (sc.finishConnect()) {
                    // 注册读事件
                    sc.register(selector, SelectionKey.OP_READ);
                    // 写入响应消息( FIXME 存在'写半包'的问题 )
                    doWrite(sc, "QUERY TIME ORDER");
                } else {
                    // 连接失败，进程退出
                    System.exit(1);
                }
            }

            // 处理可读消息
            if (key.isReadable()) {
                // 读取服务端消息
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);
                if (readBytes > 0) {
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    String body = new String(bytes, StandardCharsets.UTF_8);
                    System.out.println("Now is: " + body);
                    this.stop = true;
                }
                if (readBytes < 0) {
                    // 对端链路已关闭，需要关闭通道释放资源
                    key.cancel();
                    sc.close();
                }
            }
        }
    }

    private void doConnect() throws IOException {
        // 4、异步连接服务端
        boolean connected = socketChannel.connect(new InetSocketAddress(host, port));
        // 5、判断是否连接成功？
        if (connected) {
            //  如果连接成功，则注册读状态到多路复用器中；
            socketChannel.register(selector, SelectionKey.OP_READ);
            // 发送请求消息
            doWrite(socketChannel, "QUERY TIME ORDER");
        } else {
            //  如果连接失败，说明客户端已经发送sync包，服务端没有返回ack包，物理链路还没有建立
            socketChannel.register(selector, SelectionKey.OP_CONNECT);
        }
    }

    private void doWrite(SocketChannel sc, String body) throws IOException {
        byte[] req = body.getBytes();
        ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
        writeBuffer.put(req);
        writeBuffer.flip();
        sc.write(writeBuffer);
        if (!writeBuffer.hasRemaining()) {
            System.out.println("Send clientRequest 2 server succeed.");
        }
    }
}
